package com.inspur.cloud.service.dataspace;

import com.inspur.cloud.entity.ambari.ClusterServicesEntity;
import com.inspur.cloud.entity.dataspace.*;
import com.inspur.cloud.entity.dataspace.common.OpEntityResult;
import com.inspur.cloud.entity.dataspace.dto.HdfsInfo;
import com.inspur.cloud.entity.dataspace.dto.ResourceServiceDTO;
import com.inspur.cloud.entity.dataspace.dto.YarnTotalDTO;
import com.inspur.cloud.entity.dataspace.vo.*;
import com.inspur.cloud.enums.PermissionEnum;
import com.inspur.cloud.enums.QuotaEnum;
import com.inspur.cloud.enums.ResourceTypeEnum;
import com.inspur.cloud.exception.ApiException;
import com.inspur.cloud.exception.YarnException;
import com.inspur.cloud.security.AuthorizationHelper;
import com.inspur.cloud.service.ambari.ClusterServicesSerivce;
import com.inspur.cloud.util.StringUtil;
import com.inspur.cloud.util.TConstants;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Service;
import org.springframework.transaction.annotation.Transactional;
import org.springframework.transaction.interceptor.TransactionAspectSupport;

import java.util.ArrayList;
import java.util.Arrays;
import java.util.Date;
import java.util.List;

@Service
public class StatusService {
    @Autowired
    HdfsResourceService hdfsResourceService;
    @Autowired
    HbaseResourceService hbaseResourceService;
    @Autowired
    HiveResourceService hiveResourceService;
    @Autowired
    YarnResourceService yarnResourceService;
    @Autowired
    UserService userService;
    @Autowired
    UserResRelationSerivce userResRelationSerivce;
    @Autowired
    HdfsService hdfsService;
    @Autowired
    RangerService rangerService;
    @Autowired
    HbaseService hbaseService;
    @Autowired
    HiveService hiveService;
    @Autowired
    YarnService yarnService;
    @Autowired
    KafkaService kafkaService;
    @Autowired
    HdfsManagerService hdfsManagerService;
    @Autowired
    YarnManagerService yarnManagerService;
    @Autowired
    LoginService loginService;
    @Autowired
    ClusterConfigUtil clusterConfigUtil;
    @Autowired
    KafkaManagerService kafkaManagerService;
    @Autowired
    HiveManagerService hiveManagerService;
    @Autowired
    KafkaResourceService kafkaResourceService;
    @Autowired
    UserResourceRelationService userResourceRelationService;
    @Autowired
    ClusterServicesSerivce clusterServicesSerivce;

    private static Logger logger = LoggerFactory.getLogger(StatusService.class);

    public OpEntityResult<StatusVO> showStatus(Integer tenantId) {
        UserEntity user = userService.findUserById(tenantId);
        OpEntityResult<StatusVO> result = new OpEntityResult();
        StatusVO vo = new StatusVO();
        List<String> resourceList = new ArrayList<>();
        HdfsResourceEntity hdfs = hdfsResourceService.findFirstByOwner(tenantId, 0);
        if (hdfs != null) {
            HdfsInfo hdfsInfo = hdfsService.getContentSummary(TConstants.USRPATH + TConstants.DIRECTORY + user.getName());
            vo.setHdfsUsed(hdfsInfo.getUsed());
            vo.setHdfsResNum(new Long(hdfsInfo.getFileAndDirectoryCount()).intValue());
            vo.setHdfsDirNum(hdfs.getFileNum());
            vo.setHdfsUnit(hdfs.getUnit());
            Float remain = hdfsInfo.getRemaining() / QuotaEnum.getSizeByType(hdfs.getUnit()) * 1.0F;
            vo.setHdfsRemaining(remain.longValue());

        }
        HbaseResourceEntity hbase = hbaseResourceService.findFirstByOwner(tenantId);
        if (hbase != null) {
            vo.setHbaseSpaceName(user.getName());
        }
        HiveResourceEntity hive = hiveResourceService.findFirstByOwner(tenantId);
        if (hive != null) {
            vo.setHbaseSpaceName(user.getName());
        }
        YarnTotalDTO yarnTotal = yarnService.getQueueMetricTenant(user.getName());
        vo.setYarnCapacity((int) yarnTotal.getCapacity());
        vo.setYarnTotalCpu(yarnTotal.getTotalCore());
        vo.setYarnUnUsedCpu(yarnTotal.getUnusedCore());
        vo.setYarnUsedCpu(yarnTotal.getUsedCore());
        vo.setYarnTotalMemory(yarnTotal.getTotalMemory());
        vo.setYarnUsedMemory(yarnTotal.getUsedMemory());
        vo.setYarnUnUsedMemory(yarnTotal.getUnusedMemory());
        vo.setYarnMemoryUnit(1);
        vo.setKafkaTopicNum(userResRelationSerivce.findByUserIdAndResourceType(tenantId, "KAFKA").size());

        vo.setResourceList(loginService.getServiceList());
        result.setData(vo);
        return result;
    }

    public OpEntityResult<StatusAllVO> all() {
        OpEntityResult<StatusAllVO> result = new OpEntityResult(1, "查询成功");
        HdfsInfo hdfsInfo = hdfsService.getHDFSStatus();
        StatusAllVO vo = new StatusAllVO();
        if (hdfsInfo != null) {
            vo.setHdfsAll(hdfsInfo.getCapacity());
            vo.setHdfsUsed(hdfsInfo.getUsed());
            vo.setHdfsUnUsed(hdfsInfo.getRemaining());
        }
        YarnTotalDTO yarnInfo = yarnService.getQueueMetric();
        vo.setCpuTotal(yarnInfo.getTotalCore());
        vo.setCpuUnUsed(yarnInfo.getUnusedCore());
        vo.setCpuUsed(yarnInfo.getUsedCore());
        vo.setMemoryTotal(yarnInfo.getTotalMemory() != null ? yarnInfo.getTotalMemory() * 1024 * 1024 : 0L);//这里可能为空
        vo.setMemoryUnUsed(yarnInfo.getUnusedMemory() != null ? yarnInfo.getUnusedMemory() * 1024 * 1024 : 0L);
        vo.setMemoryUsed(yarnInfo.getUsedMemory() != null ? yarnInfo.getUsedMemory() * 1024 * 1024 : 0L);

        result.setData(vo);
        return result;
    }

    @Transactional(transactionManager = "transactionManagerDataspace")
    public OpEntityResult<ResourceRetVO> add(ResourceServiceReq resourceServiceReq) {
        Date now = new Date();
        OpEntityResult<ResourceRetVO> ret = new OpEntityResult<>(1, "");
        ResourceRetVO vo = new ResourceRetVO();
        UserEntity user = userService.findUserById(resourceServiceReq.getUid());
        if (user == null) {
            return OpEntityResult.fail("UID对应的用户不存在");
        }
        UserEntity currentUser = AuthorizationHelper.getUser();
        List<ResourceServiceDTO> services = resourceServiceReq.getServices();
        if (services == null || services.size() == 0) {
            return ret;
        }
        ResourceServiceDTO resourceServiceDTO = services.get(0);
        String errMsg="";

        boolean userInRange=rangerService.isExistUser(user.getName());

        if(!userInRange ){
            //ranger里还没有用户
            try {
                rangerService.rangerUserAddByHttp(user.getName(),user.getName());
                Thread.sleep(1000);
                boolean userInRangeNow=rangerService.isExistUser(user.getName());
                if(!userInRangeNow){
                    Thread.sleep(2000);//再尝试一次
                }
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
        }
        if(!userInRange){
            //最后尝试一次
            rangerService.rangerUserAddByHttp(user.getName(),user.getName());
        }
        if (resourceServiceDTO.getHdfs() != null && isServiceExist("HDFS")) {
            if (resourceServiceDTO.getHdfs().getFileNum() == null || resourceServiceDTO.getHdfs().getFileNum() <= 0) {
                return OpEntityResult.fail("文件数输入不合法");
            }
            if (resourceServiceDTO.getHdfs().getUnit() == null) {
                return OpEntityResult.fail("配额单位（M/G/T）必传");
            }
            if (resourceServiceDTO.getHdfs().getCapacity() == null || resourceServiceDTO.getHdfs().getCapacity() <= 0) {
                return OpEntityResult.fail("配额大小输入不合法");
            }
            String path = TConstants.USRPATH + "/" + user.getName().toLowerCase();
            HdfsResourceEntity hdfsExist = hdfsResourceService.findFirstByUrl(path);
            if (hdfsExist == null) {
                HdfsResourceEntity hdfs = resourceServiceDTO.getHdfs();
                hdfs.setUrl(path);
                hdfs.setOwner(user.getId());
                hdfs.setParentId(0);
                hdfs.setStatus(1);
                hdfs.setLevel(1);
                hdfs.setUpdDate(now);
                hdfs.setInsDate(now);
                hdfs.setCreator(currentUser.getId());

                Long unUsed = hdfsManagerService.remain(0);
                Long quota = hdfs.getCapacity() * QuotaEnum.getSizeByType(hdfs.getUnit());
                if (unUsed < quota) {
                    return OpEntityResult.fail("剩余空间不足分配");
                }
                OpEntityResult<Integer> remainFileNum = hdfsManagerService.remainFileNum(0);
                Integer unUsedFileNum = remainFileNum.getData();
                Integer quotaFileNum = hdfs.getFileNum();
                if (unUsedFileNum < quotaFileNum) {
                    return OpEntityResult.fail("剩余文件数不足分配");
                }
                //调接口，然后入库 测试
                try {
                    hdfsService.makeDir(path);
                    hdfsService.setOwner(path, user.getName(), user.getName());
                    rangerService.setPermissionInRanger(ResourceTypeEnum.HDFS.name().toLowerCase(), path, PermissionEnum.getAllPermissionByType(ResourceTypeEnum.HDFS.getType()), user.getName());
                    hdfsService.setQuota(path, quota, hdfs.getFileNum());
                    hdfsResourceService.save(hdfs);
                    UserResourceRelationEntity ur = new UserResourceRelationEntity();
                    ur.setResourceId(hdfs.getId());
                    ur.setInsDate(now);
                    ur.setUpdDate(now);
                    ur.setStatus(1);
                    ur.setUserId(user.getId());
                    ur.setResourceType(ResourceTypeEnum.HDFS.name());
                    ur.setResourceContent(hdfs.getUrl());
                    ur.setAuthor(PermissionEnum.getAllPermissionByType(ResourceTypeEnum.HDFS.getType()));
                    ur.setResourceLevel(1);
                    userResRelationSerivce.save(ur);

                    user.setUpdDate(now);//用户创建了资源， 要更新用户的时间
                    userService.save(user);
                } catch (Exception e) {//开始回滚
                    logger.error("创建hdfs目录时出错", e);
                    errMsg=errMsg+"创建hdfs目录时出错; ";
                    Integer primaryId=hdfs.getId();
                    if(hdfs.getId()==null){
                        hdfsExist=hdfsResourceService.findFirstByUrl(path);
                        if(hdfsExist!=null) primaryId=hdfsExist.getId();
                    }
                    if(primaryId!=null){
                        List<Integer> idList=new ArrayList<>();
                        idList.add(primaryId);
                        hdfsResourceService.deleteByIdList(idList);
                        userResRelationSerivce.delByResourceId(primaryId,ResourceTypeEnum.HDFS.getResourceType());
                        try{
                            rangerService.setPermissionInRanger(ResourceTypeEnum.HDFS.name().toLowerCase(), path, "", user.getName());
                        }catch (Exception re){
                            logger.error("hdfs reset ranger error.",e);
                        }
                    }
                }
                vo.setHdfsId(null);//返回HDFS id
            } else {
                vo.setHdfsId(hdfsExist.getId());
            }
        }

        if (resourceServiceDTO.getHbase() != null && isServiceExist("HBASE")) {
            HbaseResourceEntity hbase = resourceServiceDTO.getHbase();
            HbaseResourceEntity hbaseExist = hbaseResourceService.findBySpaceName(user.getName());
            if (hbaseExist != null) {
                vo.setHbaseId(hbaseExist.getId());
                return OpEntityResult.fail("该Hbase命名空间已被占用/保留，请联系管理员");
            } else {
                hbase.setSpaceName(user.getName());
                hbase.setOwner(user.getId());
                hbase.setParentId(0);
                hbase.setStatus(1);
                hbase.setLevel(1);
                hbase.setUpdDate(now);
                hbase.setInsDate(now);
                hbase.setCreator(currentUser.getId());
                try {
                    hbaseResourceService.save(hbase);
                    hbaseService.createNameSpace(user.getName());
                    String resource = hbase.getSpaceName() + ":*" + "|*|*";
                    rangerService.setPermissionInRanger(ResourceTypeEnum.HBASE.name().toLowerCase(), resource, PermissionEnum.getAllPermissionByType(ResourceTypeEnum.HBASE.getType()), user.getName());

                    UserResourceRelationEntity ur = new UserResourceRelationEntity();
                    ur.setResourceId(hbase.getId());
                    ur.setInsDate(now);
                    ur.setUpdDate(now);
                    ur.setStatus(1);
                    ur.setUserId(user.getId());
                    ur.setResourceType(ResourceTypeEnum.HBASE.name());
                    ur.setResourceContent(hbase.getSpaceName());
                    ur.setAuthor(PermissionEnum.getAllPermissionByType(ResourceTypeEnum.HBASE.getType()));
                    ur.setResourceLevel(1);
                    userResRelationSerivce.save(ur);
                    vo.setHbaseId(hbase.getId());

                    user.setUpdDate(now);//用户创建了资源， 要更新用户的时间
                    userService.save(user);
                } catch (Exception e) {
                    logger.error("创建Hbase数据库时出错,maybe recall,rollback", e);
                    errMsg=errMsg+"创建Hbase数据库时出错; ";
                    try {
                        if(hbase.getId()!=null) {
                            hbaseResourceService.deleteById(hbase.getId());
                            userResRelationSerivce.delByResourceId(hbase.getId(),ResourceTypeEnum.HBASE.getResourceType());
                        }else{
                            hbaseExist=hbaseResourceService.findBySpaceName(user.getName());
                            if(hbaseExist!=null){
                                hbaseResourceService.deleteById(hbaseExist.getId());
                                userResRelationSerivce.delByResourceId(hbaseExist.getId(),ResourceTypeEnum.HBASE.getResourceType());
                            }
                        }
                        //hbaseService.deleteNameSpace(user.getName());//回滚此处 都失敗了，不回滾
                        String resource = hbase.getSpaceName() + ":*" + "|*|*";
                        rangerService.setPermissionInRanger(ResourceTypeEnum.HBASE.name().toLowerCase(), resource, "", user.getName());
                        vo.setHbaseId(null);
                    } catch (Exception ee) {
                        logger.error("hbase delete error...", e);
                    }
                    vo.setHbaseId(null);
                }
            }
        }

        if (resourceServiceDTO.getHive() != null && isServiceExist("HIVE")) {
            HiveAddVO hiveAddVO = new HiveAddVO();
            hiveAddVO.setDatabaseName(user.getName());
            HiveResourceEntity hiveExist = hiveResourceService.findByDatabaseName(user.getName());
            if (hiveExist != null) {
                vo.setHiveId(hiveExist.getId());
                return OpEntityResult.fail("该Hive数据库已被占用/保留，请联系管理员");
            } else {
                try {
                    OpEntityResult<Integer> hiveAddResult = hiveManagerService.addRes(hiveAddVO, user);
                    if (hiveAddResult != null && hiveAddResult.getCode() == 1 && hiveAddResult.getData() != null) {
                        vo.setHiveId(hiveAddResult.getData());
                    }
                    user.setUpdDate(now);
                    userService.save(user);
                } catch (Exception e) {
                    logger.error("create Hive error,maybe recall,rollback", e);
                    if(vo.getHiveId()!=null) {
                        hiveResourceService.deleteById(vo.getHiveId());
                        userResRelationSerivce.delByResourceId(vo.getHiveId(),ResourceTypeEnum.HIVE.getResourceType());
                        logger.error("me rollback -----------------------{} @",user.getName());
                    }else{
                        HiveResourceEntity hive=hiveResourceService.findByDatabaseName(user.getName());
                        if(hive!=null){
                            hiveResourceService.deleteById(hive.getId());
                            userResRelationSerivce.delByResourceId(hive.getId(),ResourceTypeEnum.HIVE.getResourceType());
                            logger.error("me rollback -----------------------{} #",user.getName());
                        }
                    }
                    errMsg=errMsg+"创建Hive数据库时出错; ";
                    vo.setHiveId(null);
                    try {
                        hiveService.dropDatabaseForce(user.getName());//回滚此处
                        String resource = user.getName() + "|*|*";
                        rangerService.setPermissionInRanger(ResourceTypeEnum.HIVE.name().toLowerCase(), resource, "", user.getName());
                    } catch (Exception ee) {
                        logger.error("hive delete error...", ee);
                    }
                }
            }
        }

        if (resourceServiceDTO.getYarn() != null && isServiceExist("YARN")) {
            if (resourceServiceDTO.getYarn().getRatio() == null) {
                return OpEntityResult.fail("yarn队列比例必传");
            }
            if (resourceServiceDTO.getYarn().getRatio().equals(0f)) {
                return OpEntityResult.fail("yarn队列占比不能是0");
            }
            YarnResourceEntity yarnExist = yarnResourceService.findFirstByQueueNameAndParentId(user.getName(), 0);
            if (yarnExist == null) {
                YarnResourceEntity yarn = resourceServiceDTO.getYarn();
                yarn.setQueueName(user.getName());
                yarn.setOwner(user.getId());
                yarn.setParentId(0);
                yarn.setStatus(1);
                yarn.setLevel(1);
                yarn.setUpdDate(now);
                yarn.setInsDate(now);
                yarn.setCreator(currentUser.getId());
                try {
                    addOrModifyYarn(user, yarn, true);
                    vo.setYarnId(yarn.getId());

                    user.setUpdDate(now);
                    userService.save(user);
                } catch (Exception e) {//回滚
                    logger.error("创建Yarn队列时出错", e);
                    errMsg=errMsg+"创建Yarn队列时出错; ";
                    Integer primaryId=yarn.getId();
                    vo.setYarnId(null);
                    if(primaryId==null){
                        yarnExist=yarnResourceService.findFirstByQueueNameAndParentId(user.getName(),0);
                        if(yarnExist!=null) primaryId=yarnExist.getId();
                    }
                    if(primaryId!=null){
                        yarnResourceService.deleteById(primaryId);
                        userResRelationSerivce.delByResourceId(primaryId,ResourceTypeEnum.YARN.getResourceType());
                        try{
                            rangerService.setPermissionInRanger(ResourceTypeEnum.YARN.name().toLowerCase(), TConstants.ROOT + "." + user.getName(), "", user.getName());
                        }catch (Exception re){
                            logger.error("yarn reset ranger error.",e);
                        }
                    }
                }

            } else {
                vo.setYarnId(yarnExist.getId());
                return OpEntityResult.fail("该队列已被占用/保留，请联系管理员");
            }

        }

        if (resourceServiceDTO.getKafka() != null && isServiceExist("KAFKA")) {
            KafkaAddVO kafkaAddVO = new KafkaAddVO();
            kafkaAddVO.setType(2);
            kafkaAddVO.setPartitionNum(resourceServiceDTO.getKafka().getPartitionNum());
            kafkaAddVO.setReplication(resourceServiceDTO.getKafka().getReplication());
            kafkaAddVO.setTopicName(user.getName());
            Integer kafkaId=null;
            try {
                OpEntityResult<Integer> kafkaAddResult = kafkaManagerService.addRes(kafkaAddVO, user);
                if (kafkaAddResult != null && kafkaAddResult.getCode() == 1 && kafkaAddResult.getData() != null) {
                    vo.setKafkaId(kafkaAddResult.getData());
                    kafkaId=kafkaAddResult.getData();
                }
                user.setUpdDate(now);
                userService.save(user);
            } catch (Exception e) {
                logger.error("创建kafka Topic 时出错", e);
                errMsg=errMsg+"创建kafka Topic时出错; ";
                vo.setKafkaId(null);
                Integer primaryId=kafkaId;
                if(primaryId==null){
                    KafkaResourceEntity kafkaExist=kafkaResourceService.findFirstByTopicName(user.getName());
                    if(kafkaExist!=null) primaryId=kafkaExist.getId();
                }
                if(primaryId!=null){
                    kafkaResourceService.deleteById(primaryId);
                    userResRelationSerivce.delByResourceId(primaryId,ResourceTypeEnum.KAFKA.getResourceType());
                    try{
                        rangerService.setPermissionInRanger(ResourceTypeEnum.KAFKA.name().toLowerCase(), user.getName(), "", user.getName());
                    }catch (Exception re){
                        logger.error("kafka reset ranger error.",e);
                    }
                }
            }
        }
        if(StringUtil.isEmpty(errMsg)){
            ret.setCode(1);
            ret.setMsg("创建资源成功");
        }else {
            ret.setCode(1);
            ret.setMsg("创建资源部分成功");
        }
        vo.setError(errMsg);
        ret.setData(vo);
        return ret;
    }

    /**
     * 判断服务是否存在
     * @param service
     * @return
     */
    public boolean isServiceExist(String service) {
        List<ClusterServicesEntity> serviceAll = clusterServicesSerivce.findAll();
        List<String> serviceList = new ArrayList<>();
        for (ClusterServicesEntity entity : serviceAll) {
            String serviceName = entity.getServiceName();
            serviceList.add(serviceName);
        }
        return serviceList.contains(service.toUpperCase());
    }
    // @Transactional

    /**
     * 只允许修改HDFS和YARN的配置
     */
    public OpEntityResult<ResourceRetVO> modify(ResourceServiceReq resourceServiceReq) {
        Date now = new Date();
        OpEntityResult<ResourceRetVO> ret = new OpEntityResult<>(1, "");
        ResourceRetVO vo = new ResourceRetVO();
        UserEntity currentUser = AuthorizationHelper.getUser();
        UserEntity user = userService.findUserById(resourceServiceReq.getUid());
        if (user == null) {
            return OpEntityResult.fail("UID对应的用户不存在");
        }
        List<ResourceServiceDTO> services = resourceServiceReq.getServices();
        if (services == null || services.size() == 0) {
            return ret;
        }
        String errMsg="";
        ResourceServiceDTO resourceServiceDTO = services.get(0);
        if (resourceServiceDTO.getHdfs() != null && resourceServiceDTO.getHdfs().getId() != null) {
            if (resourceServiceDTO.getHdfs().getFileNum() == null || resourceServiceDTO.getHdfs().getFileNum() <= 1) {
                return OpEntityResult.fail("文件数输入不合法");
            }
            if (resourceServiceDTO.getHdfs().getCapacity() == null || resourceServiceDTO.getHdfs().getCapacity() <= 0) {
                return OpEntityResult.fail("配额输入不合法");
            }
            if (resourceServiceDTO.getHdfs().getUnit() == null) {
                return OpEntityResult.fail("配额单位（M/G/T）必传");
            }
            HdfsResourceEntity hdfsVO = resourceServiceDTO.getHdfs();
            HdfsResourceEntity hdfs = hdfsResourceService.findHdfsResourceEntityById(hdfsVO.getId());
            if (hdfs == null) {
                return OpEntityResult.fail("HDFS资源ID对应的路径数据库中不存在");
            }
            hdfs.setUpdDate(now);
            //调接口，然后入库 测试
            hdfs.setFileNum(resourceServiceDTO.getHdfs().getFileNum());
            hdfs.setCapacity(resourceServiceDTO.getHdfs().getCapacity());
            hdfs.setUnit(resourceServiceDTO.getHdfs().getUnit());
            Long quota = hdfs.getCapacity() * QuotaEnum.getSizeByType(hdfs.getUnit());
            hdfsService.setQuota(hdfs.getUrl(), quota, hdfs.getFileNum());
            hdfsResourceService.save(hdfs);

            user.setUpdDate(now);//用户创建了资源， 要更新用户的时间
            userService.save(user);

            vo.setHdfsId(hdfs.getId());//返回HDFS id
//            ret.setMsg(ret.getMsg()+"  修改HDFS资源成功 ");
        } else if (resourceServiceDTO.getHdfs() != null && resourceServiceDTO.getHdfs().getId() == null) {//新增
            ResourceServiceReq req = new ResourceServiceReq();
            List<ResourceServiceDTO> servicesList = new ArrayList<>();
            req.setUid(user.getId());
            ResourceServiceDTO resourceService = new ResourceServiceDTO();
            resourceService.setHdfs(resourceServiceDTO.getHdfs());
            servicesList.add(resourceService);
            req.setServices(servicesList);
            OpEntityResult<ResourceRetVO> retVO = add(req);
            if (retVO != null && retVO.getCode() == 1 && retVO.getData() != null) {
                vo.setHdfsId(retVO.getData().getHdfsId());
            }
            if (retVO != null){
                errMsg=errMsg+retVO.getData().getError();
            }
        }

        if (resourceServiceDTO.getHbase() != null && resourceServiceDTO.getHbase().getId() == null) {
            ResourceServiceReq req = new ResourceServiceReq();
            req.setUid(user.getId());
            List<ResourceServiceDTO> servicesList = new ArrayList<>();
            ResourceServiceDTO resourceService = new ResourceServiceDTO();
            resourceService.setHbase(resourceServiceDTO.getHbase());
            servicesList.add(resourceService);
            req.setServices(servicesList);
            OpEntityResult<ResourceRetVO> retVO = add(req);
            if (retVO != null && retVO.getCode() == 1 && retVO.getData() != null) {
                vo.setHbaseId(retVO.getData().getHbaseId());
            }
            if (retVO != null){
//                ret.setMsg(ret.getMsg()+retVO.getMsg());
                errMsg=errMsg+retVO.getData().getError();
            }
        }

        if (resourceServiceDTO.getHive() != null && resourceServiceDTO.getHive().getId() == null) {
            ResourceServiceReq req = new ResourceServiceReq();
            req.setUid(user.getId());
            List<ResourceServiceDTO> servicesList = new ArrayList<>();
            ResourceServiceDTO resourceService = new ResourceServiceDTO();
            resourceService.setHive(resourceServiceDTO.getHive());
            servicesList.add(resourceService);
            req.setServices(servicesList);
            OpEntityResult<ResourceRetVO> retVO = add(req);
            if (retVO != null && retVO.getCode() == 1 && retVO.getData() != null) {
                vo.setHiveId(retVO.getData().getHiveId());
            }
            if (retVO != null){
//                ret.setMsg(ret.getMsg()+retVO.getMsg());
                errMsg=errMsg+retVO.getData().getError();
            }
        }

        if (resourceServiceDTO.getYarn() != null && resourceServiceDTO.getYarn().getId() != null) {
            //编辑
            if (resourceServiceDTO.getYarn().getRatio() == null || resourceServiceDTO.getYarn().getRatio() <= 0f ||
                    resourceServiceDTO.getYarn().getRatio() > 100f) {
                return OpEntityResult.fail("yarn比例输入不合法");
            }
            YarnResourceEntity yarnExist = yarnResourceService.findYarnResourceEntityById(resourceServiceDTO.getYarn().getId());
            if (yarnExist == null) {
                return OpEntityResult.fail("yarn资源不存在");
            }
            if (yarnExist.getRatio() != resourceServiceDTO.getYarn().getRatio()) {//如果并没有更改yarn占比，就不去调接口。这里不会有问题，因为还有个同步功能同步最新的数据
                addOrModifyYarn(user, resourceServiceDTO.getYarn(), false);//这里新增和编辑其实是同一个接口
                user.setUpdDate(now);
                userService.save(user);
            }
            vo.setYarnId(resourceServiceDTO.getYarn().getId());
//            ret.setMsg(ret.getMsg()+" 修改YARN资源成功 ");
        } else if (resourceServiceDTO.getYarn() != null && resourceServiceDTO.getYarn().getId() == null) {
            //新建
            ResourceServiceReq req = new ResourceServiceReq();
            req.setUid(user.getId());
            List<ResourceServiceDTO> servicesList = new ArrayList<>();
            ResourceServiceDTO resourceService = new ResourceServiceDTO();
            resourceService.setYarn(resourceServiceDTO.getYarn());
            servicesList.add(resourceService);
            req.setServices(servicesList);
            OpEntityResult<ResourceRetVO> retVO = add(req);
            if (retVO != null && retVO.getCode() == 1 && retVO.getData() != null) {
                vo.setYarnId(retVO.getData().getYarnId());
            }
            if (retVO != null){
//                ret.setMsg(ret.getMsg()+retVO.getMsg());
                errMsg=errMsg+retVO.getData().getError();
            }
        }

        vo.setError(errMsg);
        if(StringUtil.isEmpty(errMsg)){
            ret.setCode(1);
            ret.setMsg("修改资源成功");
        }else {
            ret.setCode(1);
            ret.setMsg("修改资源部分成功");
        }
        ret.setData(vo);
        return ret;
    }

    public OpEntityResult<List<ResourceServiceDTO>> one(Integer tenantId) {
        OpEntityResult<List<ResourceServiceDTO>> ret = new OpEntityResult<>(1, "查询资源成功");
        if (tenantId == null) {
            if (AuthorizationHelper.isTenant()) {
                tenantId = AuthorizationHelper.getUser().getId();
            } else {
                return new OpEntityResult<>(-1, "租户ID为空");
            }
        }
        ResourceServiceDTO resourceServiceDTO = new ResourceServiceDTO();

        HdfsResourceEntity hdfs = hdfsResourceService.findFirstByOwner(tenantId, 0);
        resourceServiceDTO.setHdfs(hdfs);
        HbaseResourceEntity hbaseResourceEntity = hbaseResourceService.findFirstByOwner(tenantId);
        resourceServiceDTO.setHbase(hbaseResourceEntity);
        HiveResourceEntity hiveResourceEntity = hiveResourceService.findFirstByOwner(tenantId);
        resourceServiceDTO.setHive(hiveResourceEntity);
        YarnResourceEntity yarnResourceEntity = yarnResourceService.findQueueByOwner(tenantId);
        resourceServiceDTO.setYarn(yarnResourceEntity);
        KafkaResourceEntity kafkaResourceEntity = new KafkaResourceEntity();
        List<UserResourceRelationEntity> urs = userResRelationSerivce.findByUserIdAndResourceType(tenantId, ResourceTypeEnum.KAFKA.name());
        if (urs != null) {
            kafkaResourceEntity.setKafkaTopicNum(urs.size());
        }
        resourceServiceDTO.setKafka(kafkaResourceEntity);
        List<ResourceServiceDTO> list = new ArrayList<>();
        list.add(resourceServiceDTO);
        ret.setData(list);
        return ret;
    }

    /**
     * @param userEntity 当前用户
     * @param yarn       前端传来的YARN
     * @param add        是否要创建子队列，true的时候是新增，false的时候是编辑
     * @return
     */
    public Integer addOrModifyYarn(UserEntity userEntity, YarnResourceEntity yarn, boolean add) {
        Date now = new Date();
        if (yarn.getRatio() <= 0f) {
            throw new YarnException("不允许创建占比为0的队列");
        }
        yarnManagerService.sync(0, 1);
        List<YarnResourceEntity> yarnList = yarnResourceService.findByParentId(0);
        YarnResourceEntity old = new YarnResourceEntity();
        if (!add) {
            old = yarnResourceService.findYarnResourceEntityById(yarn.getId());//获取数据库里的老队列，这里肯定不为空
        }
        if (yarnList == null || yarnList.size() == 0) {
            throw new YarnException("获取不到yarn队列初始化信息，请先联系管理员同步");
        }
        String powerStr = PermissionEnum.getAllPermissionByType(ResourceTypeEnum.YARN.getType());
        Float remainRatio;//default新额度
        Float ratio = 0.0F;//default的原始额度
        for (YarnResourceEntity yarnResourceEntity : yarnList) {
            if (yarnResourceEntity.getQueueName().equals(TConstants.YARN_DEFAULT)) {
                ratio = yarnResourceEntity.getRatio();//获得default原始额度
                break;
            }
        }
        Float maxRatio = ratio;
        if (!add) {
            maxRatio = maxRatio + old.getRatio();//能设置的最大额度=已有的+default
        }
        if (yarn.getRatio() >= maxRatio) {//前端传值大于能设置的最大值
            throw new YarnException("请调整当前队列占比，使其小于剩余可用额度");
        } else {
            remainRatio = maxRatio - yarn.getRatio();
//            if (add) {
//                remainRatio = maxRatio - yarn.getRatio();//给default的新额度 原始-前端
//            } else {
//                remainRatio = maxRatio - yarn.getRatio();//原始default+旧的额度-传值
//            }
        }
        boolean contain = false;//防止重复创建
        for (YarnResourceEntity yarnResourceEntity : yarnList) {
            if (yarnResourceEntity.getQueueName().equals(yarn.getQueueName())) {
                contain = true;
                break;
            }
        }
        if (remainRatio < 1.0f) {
            throw new YarnException("default额度不能小于1%");
        }

        if (!contain) {
            yarnList.add(yarn);//加入新增的这个队列
        }
        Integer size = yarnList.size();
        String[] queueArr = new String[size];
        String[] ratioArr = new String[size];
        for (int i = 0; i < size; i++) {
            YarnResourceEntity yarnInfo = yarnList.get(i);
            queueArr[i] = yarnInfo.getQueueName();
            if (yarnInfo.getQueueName().equals(TConstants.YARN_DEFAULT)) {
                if(remainRatio>100f){
                    remainRatio=100.0f;//这可不能大于100
                }
                ratioArr[i] = String.format("%f", remainRatio);
            } else {
                if (yarnInfo.getQueueName().equals(yarn.getQueueName())) {
                    ratioArr[i] = String.format("%f", yarn.getRatio());
                } else {
                    ratioArr[i] = String.format("%f", yarnInfo.getRatio());
                }
            }
        }
        yarnService.editYarnQueue(TConstants.ROOT, queueArr, ratioArr);
        if (yarn.getId() == null) {//新增和编辑使用
            YarnResourceEntity yarnGlobal = yarnResourceService.findFirstByQueueName(yarn.getQueueName());
            if (yarnGlobal != null) {
                throw new ApiException("全局yarn队列中不允许存在同名队列");
            }

            yarnResourceService.save(yarn);
            UserResourceRelationEntity ur = new UserResourceRelationEntity();
            ur.setResourceId(yarn.getId());
            ur.setInsDate(now);
            ur.setUpdDate(now);
            ur.setStatus(1);
            ur.setUserId(userEntity.getId());
            ur.setResourceType(ResourceTypeEnum.YARN.name());
            ur.setResourceContent(yarn.getQueueName());
            ur.setAuthor(PermissionEnum.getAllPermissionByType(ResourceTypeEnum.YARN.getType()));
            ur.setResourceLevel(1);
            userResRelationSerivce.save(ur);
            String resource = TConstants.ROOT + "." + userEntity.getName();
            rangerService.setPermissionInRanger(ResourceTypeEnum.YARN.name().toLowerCase(), resource, powerStr, userEntity.getName());

            YarnResourceEntity defaultYarn = yarnResourceService.findFirstByQueueNameAndParentId(TConstants.YARN_DEFAULT, 0);
            defaultYarn.setRatio(remainRatio);
            defaultYarn.setUpdDate(now);
            yarnResourceService.save(defaultYarn);

        } else {
            old.setRatio(yarn.getRatio());
            yarnResourceService.save(old);

            YarnResourceEntity defaultYarn = yarnResourceService.findFirstByQueueNameAndParentId(TConstants.YARN_DEFAULT, 0);
            defaultYarn.setRatio(remainRatio);
            defaultYarn.setUpdDate(now);
            yarnResourceService.save(defaultYarn);
        }

//        ResourceCommonDTO resourceCommonDTO=userResourceRelationService.getResourceCommon(yarn.getId(),"YARN");
//        userResourceRelationService.addUserRelation(resourceCommonDTO,AuthorizationHelper.getUser(),"YARN",PermissionEnum.YARN_ADMIN.getPermission());
        return yarn.getId();
    }
}
